package com.atguigu.realtime.func;

import com.alibaba.fastjson.JSONObject;
import com.atguigu.realtime.beans.BaseDbTableProcess;
import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.util.Collector;

import java.sql.*;
import java.util.*;

/**
 * @author: 洛尘
 * @since: 2023-10-08 19:59
 * @description: 处理需要动态分流的事实表
 **/
public class BaseDbTableProcessFunction extends BroadcastProcessFunction<JSONObject, String, JSONObject> {

    private MapStateDescriptor<String, BaseDbTableProcess> mapStateDescriptor;
    //*
    private Map<String,BaseDbTableProcess> configMap = new HashMap<>();

    public BaseDbTableProcessFunction(MapStateDescriptor<String, BaseDbTableProcess> mapStateDescriptor) {
        this.mapStateDescriptor = mapStateDescriptor;
    }

    //为了控制广播流的数据在主流的数据之后到达进行如下操作 *

    //*
    @Override
    public void open(Configuration parameters) throws Exception {
        //将配置表中的配置信息加载到程序中
        //注册驱动
        Class.forName("com.mysql.cj.jdbc.Driver");
        //建立连接
        Connection connection = DriverManager.getConnection("jdbc:mysql://hadoop105:3306/gmall0417_config?" +
                "user=root&password=000000&useUnicode=true&" +
                "characterEncoding=utf8&serverTimeZone=Asia/Shanghai&useSSL=false");

        //定义sql
        String sql="select * from gmall0417_config.table_process_dwd";

        //获取数据库操作对象
        PreparedStatement ps = connection.prepareStatement(sql);
        //执行sql语句
        ResultSet rs = ps.executeQuery();
        ResultSetMetaData metaData = rs.getMetaData();
        //处理结果集
        while (rs.next()){
            JSONObject jsonObj = new JSONObject();
            for (int i = 1; i <= metaData.getColumnCount(); i++) {
                String columnName = metaData.getColumnName(i);
                Object columnValue = rs.getObject(i);
                jsonObj.put(columnName,columnValue);
            }
            BaseDbTableProcess tableProcess = jsonObj.toJavaObject(BaseDbTableProcess.class);
            String sourceTable = tableProcess.getSourceTable();
            String sourceType = tableProcess.getSourceType();
            String key=sourceTable+":"+sourceType;
            configMap.put(key,tableProcess);
        }
        //释放资源
        rs.close();
        ps.close();
        connection.close();

    }

    //处理主流业务数据
    @Override
    public void processElement(JSONObject jsonObject, ReadOnlyContext ctx, Collector<JSONObject> collector) throws Exception {
        //获取表名
        String table = jsonObject.getString("table");
        //获取操作类型
        String type = jsonObject.getString("type");
        //拼接key
        String key=table+":"+type;

        //获取广播状态
        ReadOnlyBroadcastState<String, BaseDbTableProcess> broadcastState = ctx.getBroadcastState(mapStateDescriptor);
        //根据key到广播状态中获取配置信息
        BaseDbTableProcess tableProcess=null;
        if ((tableProcess=broadcastState.get(key))!=null||((tableProcess=configMap.get(key))!=null)){
            //如果配置信息不为空，说明是需要动态分流处理的事实表数据，发送到下游
            JSONObject dataJsonObj = jsonObject.getJSONObject("data");
            //向下游传递数据前，过滤掉不需要传递的属性
            String sinkColumns = tableProcess.getSinkColumns();
            filterColumns(dataJsonObj,sinkColumns);
            //在向下游传递数据前，补充将要输出到的kafka主题
            String sinkTable = tableProcess.getSinkTable();
            dataJsonObj.put("sink_table" ,sinkTable);
            //在向下游传递数据前，补充事件时间
            Long ts = jsonObject.getLong("ts");
            dataJsonObj.put("ts",ts);
            collector.collect(dataJsonObj);
        }
    }

    private void filterColumns(JSONObject dataJsonObj, String sinkColumns) {
        String[] columnArr = sinkColumns.split(",");
        List<String> columnList = Arrays.asList(columnArr);
        Set<Map.Entry<String, Object>> entries = dataJsonObj.entrySet();
        entries.removeIf(extry->!columnList.contains(extry.getKey()));
    }

    //处理广播流业务数据方法
    @Override
    public void processBroadcastElement(String jsonStr, Context context, Collector<JSONObject> collector) throws Exception {

        BroadcastState<String, BaseDbTableProcess> broadcastState = context.getBroadcastState(mapStateDescriptor);
        JSONObject jsonObj = JSONObject.parseObject(jsonStr);
        if ("d".equals(jsonObj.getString("op"))){
            //如果是删除操作，将对应的配置信息进行删除
            BaseDbTableProcess before = jsonObj.getObject("before", BaseDbTableProcess.class);
            String sourceTable = before.getSourceTable();
            String sourceType = before.getSourceType();
            String key=sourceTable+":"+sourceType;
            broadcastState.remove(key);
            configMap.remove(key);
        }else {
            //其它操作将对应的配置信息放到广播状态中
            BaseDbTableProcess after = jsonObj.getObject("after", BaseDbTableProcess.class);
            String sourceTable = after.getSourceTable();
            String sourceType = after.getSourceType();
            String key=sourceTable+":"+sourceType;
            broadcastState.put(key,after);
            configMap.put(key,after);
        }
    }
}